package com.shiguiwu.springboot3.rockectmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @description: 生产者
 * @author: stone
 * @date: Created by 2023/11/10 19:34
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.message.rockectmq.producer
 */
@Slf4j
public class ProducerTests {

    /**
     * 测试同步消息
     * @throws Exception
     */
    @Test
    public void testSyncProducer() throws Exception {
        log.info("啊啊哈");
        // 初始化一个producer并设置Producer group name
        DefaultMQProducer producer = new DefaultMQProducer("shiguiwuaaaaaa"); //（1）
        // 设置NameServer地址
        producer.setNamesrvAddr("192.168.192.130:9876");  //（2）
        // 启动producer
        producer.start();
        for (int i = 0; i < 99; i++) {
            // 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
            Message msg = new Message("string-topic" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ shi" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );   //（3）
            // 利用producer进行发送，并同步等待发送结果
            SendResult sendResult = producer.send(msg);   //（4）
            log.error("返回结构===> {}", sendResult);
        }
        // 一旦producer不再使用，关闭producer
        producer.shutdown();
    }

    /**
     * 测试异步消息
     */
    @Test
    public void testAsyncProducer() throws Exception {
        // 初始化一个producer并设置Producer group name
        DefaultMQProducer producer = new DefaultMQProducer("shiguiwuAA111");
        // 设置NameServer地址
        producer.setNamesrvAddr("192.168.192.130:9876");
        // 启动producer
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        int messageCount = 100;
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                // 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
                Message msg = new Message("string-topic",
                        "TagA", "keya",
                        "Hello world aaaa".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 异步发送消息, 发送结果通过callback返回给客户端
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.error("消息成功回调 ==> 序号{}发送结果消息id ={}", index,
                                sendResult.getMsgId());
                        countDownLatch.countDown();
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("异常了 ={} Exception {}}", index, e);
                        countDownLatch.countDown();
                    }
                });
            } catch (Exception e) {
                log.error("异常了 ={} Exception {}", Thread.currentThread().getName(), e);
                countDownLatch.countDown();
            }
        }
        //异步发送，如果要求可靠传输，必须要等回调接口返回明确结果后才能结束逻辑，否则立即关闭Producer可能导致部分消息尚未传输成功
        countDownLatch.await(5, TimeUnit.SECONDS);
        // 一旦producer不再使用，关闭producer
        producer.shutdown();
    }

    /**
     * 测试单向消息，不需要回调信息
     */
    @Test
    public void testOnewayProducer() throws Exception {
// 初始化一个producer并设置Producer group name
        DefaultMQProducer producer = new DefaultMQProducer("please_999_group_name");
        // 设置NameServer地址
        producer.setNamesrvAddr("192.168.192.130:9876");
        // 启动producer
        producer.start();
        for (int i = 0; i < 100; i++) {
            // 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
            Message msg = new Message("string-topic" /* Topic */,
                    "TagA" /* Tag */,
                    "OnewayProducer",
                    ("Hello RocketMQ OnewayProducer " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // 由于在oneway方式发送消息时没有请求应答处理，如果出现消息发送失败，则会因为没有重试而导致数据丢失。若数据不可丢，建议选用可靠同步或可靠异步发送方式。
            producer.sendOneway(msg);
        }
        // 一旦producer不再使用，关闭producer
        producer.shutdown();
    }


}
